package com.bw.yk01;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import scala.Tuple7;
import scala.Tuple8;

import java.util.Properties;

/*
15> {"baseProvince":{"area_code":"620000","id":"19","iso_3166_2":"CN-GS","iso_code":"CN-62","name":"甘肃","region_id":"7"},
"orderDetail":{"create_time":"2020-06-15 23:34:25","id":"13171","img_url":"http://47.93.148.192:8080/group1/M00/00/01/rBHu8l-rfvmAIpgZAAIvrX6L9fo612.jpg","order_id":"4922","order_price":"5999.00","sku_id":"1","sku_name":"小米10 至尊纪念版 双模5G 骁龙865 120HZ高刷新率 120倍长焦镜头 120W快充 8GB+128GB 陶瓷黑 游戏手机","sku_num":"1","source_id":"12","source_type":"2402","split_total_amount":"5999.00"},
"orderInfo":{"consignee":"司空有坚","consignee_tel":"13576266678","create_time":"2020-06-15 23:34:25","delivery_address":"第14大街第19号楼5单元192门","expire_time":"2020-06-15 23:49:25","id":"4922","img_url":"http://img.gmall.com/398818.jpg","operate_time":"2020-06-15 23:34:26","order_comment":"描述432833","order_status":"1004","out_trade_no":"712257226924328","province_id":"19","total_amount":"14214.00","trade_body":"华为智慧屏V55i-J 55英寸 HEGE-550B 4K全面屏智能电视机 多方视频通话 AI升降摄像头 银钻灰 京品家电等6件商品","user_id":"460"},
"skuInfo":{"category3_id":"61","create_time":"2020-11-11 22:07:35","id":"1","is_sale":"1","price":"5999","sku_default_img":"http://47.93.148.192:8080/group1/M00/00/01/rBHu8l-rfvmAIpgZAAIvrX6L9fo612.jpg","sku_desc":"小米10 至尊纪念版 双模5G 骁龙865 120HZ高刷新率 120倍长焦镜头 120W快充 12GB+256GB 陶瓷黑 游戏手机","sku_name":"小米10 至尊纪念版 双模5G 骁龙865 120HZ高刷新率 120倍长焦镜头 120W快充 8GB+128GB 陶瓷黑 游戏手机","spu_id":"1","tm_id":"1","weight":"1.00"},
"userInfo":{"birthday":"-320","create_time":"2020-06-15 23:34:19","email":"idlv7o09t@googlemail.com","gender":"F","id":"460","login_name":"idlv7o09t","name":"贺馨艺","nick_name":"馨艺","phone_num":"13359459244","user_level":"3"}}

 */

public class FlinkTM5_1 {
    public static void main(String[] args) throws Exception {
        //创建流处理表环境
        StreamExecutionEnvironment fsEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        fsEnv.setParallelism(1);
        //从dwd_trade_orders读取数据
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop-single:9092");
        properties.setProperty("group.id", "group1");
        DataStream<String> stream = fsEnv.addSource(new FlinkKafkaConsumer<>("dwd_trade_orders", new SimpleStringSchema(), properties));

        //提取对应字段转换成表 订单id,订单创建时间，省份id,省份名称，sku_id,sku_name,sku_num
        SingleOutputStreamOperator<Tuple8<String,String,String,String,String,String,String,Integer>> tDS = stream.map(new MapFunction<String, Tuple8<String,String,String,String,String,String,String,Integer>>() {
            @Override
            public Tuple8<String,String,String,String,String,String,String,Integer> map(String s) throws Exception {
                    JSONObject jo = JSON.parseObject(s);
                    return new Tuple8<String,String,String,String,String,String,String,Integer>(jo.getJSONObject("orderInfo").getString("id"),
                            jo.getJSONObject("orderInfo").getString("create_time"),
                            jo.getJSONObject("baseProvince").getString("id"),
                            jo.getJSONObject("baseProvince").getString("name"),
                            jo.getJSONObject("skuInfo").getString("id"),
                            jo.getJSONObject("skuInfo").getString("sku_name"),
                            jo.getJSONObject("userInfo").getString("id"),
                            Integer.valueOf(jo.getJSONObject("orderDetail").getString("sku_num")))
                            ;
            }
        });

        tDS.print();

        tDS.writeAsCsv("datas/tb");
//        tDS.writeAsText("datas/tb");

        fsEnv.execute();


    }
}
